/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

 

package com.sui.bigdata.rtcadmin.util;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Maps;
import com.sui.bigdata.rtcadmin.repository.model.Meta;
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * 解析创建表结构sql
 * Date: 2019/07/10
 * @author yong_chen@sui.com
 */

public class CreateTableParser {

    private static final char SQL_DELIMITER = ';';

    private static final String PATTERN_STR = "(?i)create\\s+table\\s+(\\S+)\\s*\\((.+)\\)\\s*with\\s*\\((.+)\\)";

    private static final Pattern PATTERN = Pattern.compile(PATTERN_STR);

    public static CreateTableParser newInstance(){
        return new CreateTableParser();
    }


    public boolean verify(String sql) {
        return PATTERN.matcher(sql).find();
    }


    public List<SqlParserResult> parseSql(String sql) {
        if (StringUtils.isBlank(sql)) {
            throw new RuntimeException("sql is not null");
        }
        sql = sql.replaceAll("--.*", "")
                .replaceAll("\r\n", " ")
                .replaceAll("\n", " ")
                .replace("\t", " ").trim();

        List<String> sqlArr = splitIgnoreQuota(sql, SQL_DELIMITER);
        List<SqlParserResult> sqlParserResults = new ArrayList<>();
        for (String childSql : sqlArr) {
            Matcher matcher = PATTERN.matcher(childSql);

            if (!verify(childSql)) {
                throw new RuntimeException("Syntax does not support.");
            }
            if(matcher.find()){
                String tableName = matcher.group(1);
                String fieldsInfoStr = matcher.group(2);
                String propsStr = matcher.group(3);
                Map<String, Object> props = parseProp(propsStr);

                SqlParserResult result = new SqlParserResult();
                result.setTableName(tableName);
                result.setFieldsInfoStr(fieldsInfoStr);
                result.setPropMap(props);
                result.setCreateTableStatement(childSql);
                sqlParserResults.add(result);
            }
        }

        return sqlParserResults;
    }

    private Map parseProp(String propsStr){
        String[] strs = propsStr.trim().split("'\\s*,");
        Map<String, Object> propMap = Maps.newHashMap();
        for(int i=0; i<strs.length; i++){
            List<String> ss = splitIgnoreQuota(strs[i], '=');
            String key = ss.get(0).trim();
            String value = ss.get(1).trim().replaceAll("'", "").trim();
            propMap.put(key, value);
        }

        return propMap;
    }

    public static class SqlParserResult{

        private String tableName;

        private String fieldsInfoStr;

        private String createTableStatement;

        private Map<String, Object> propMap;

        public String getTableName() {
            return tableName;
        }

        public void setTableName(String tableName) {
            this.tableName = tableName;
        }

        public String getFieldsInfoStr() {
            return fieldsInfoStr;
        }

        public void setFieldsInfoStr(String fieldsInfoStr) {
            this.fieldsInfoStr = fieldsInfoStr;
        }

        public Map<String, Object> getPropMap() {
            return propMap;
        }

        public void setPropMap(Map<String, Object> propMap) {
            this.propMap = propMap;
        }

        public String getCreateTableStatement() {
            return createTableStatement;
        }

        public void setCreateTableStatement(String createTableStatement) {
            this.createTableStatement = createTableStatement;
        }
    }

    /**
     * Split the specified string delimiter --- ignored quotes delimiter
     * @param str
     * @param delimiter
     * @return
     */
    public static List<String> splitIgnoreQuota(String str, char delimiter){
        List<String> tokensList = new ArrayList<>();
        boolean inQuotes = false;
        boolean inSingleQuotes = false;
        StringBuilder b = new StringBuilder();
        for (char c : str.toCharArray()) {
            if(c == delimiter){
                if (inQuotes) {
                    b.append(c);
                } else if(inSingleQuotes){
                    b.append(c);
                }else {
                    tokensList.add(b.toString());
                    b = new StringBuilder();
                }
            }else if(c == '\"'){
                inQuotes = !inQuotes;
                b.append(c);
            }else if(c == '\''){
                inSingleQuotes = !inSingleQuotes;
                b.append(c);
            }else{
                b.append(c);
            }
        }

        tokensList.add(b.toString());

        return tokensList;
    }

    public static void main(String[] str){
        String sql = "create table test_kudu(\n" +
                "                        id BIGINT,\n" +
                "                        name STRING,\n" +
                "                        PRIMARY KEY (id) NOT ENFORCED\n" +
                "                        )with(\n" +
                "                        'connector' = 'kudu',\n" +
                "                        'table-name' = 'impala::default.test_kudu',\n" +
                "                        'kudu-masters' = '10.201.5.195:7051,10.201.5.196:7051,10.201.5.215:7051',\n" +
                "                        'lookup.cache.max-rows' = '2',\n" +
                "                        'lookup.cache.ttl' = '600000',\n" +
                "                        'async-lookup' = 'true'\n" +
                "                        )";
        String sql2 = "CREATE TABLE hbase_dim_table_async (\n" +
                "  rowkey STRING,\n" +
                "  cf ROW < location STRING,gdp STRING >,\n" +
                "  cf2 ROW < aaa2 STRING,bbb2 STRING >,\n" +
                "  cf3 ROW < ccc3 STRING >,\n" +
                "  PRIMARY KEY (rowkey) NOT ENFORCED\n" +
                ") WITH (\n" +
                "  'connector' = 'hbase-2.2',\n" +
                "  'table-name' = 'htw_test',\n" +
                "  'lookup.async' = 'true',\n" +
                "  'lookup.cache.max-rows' = '2',\n" +
                "  'lookup.cache.ttl' = '6000000',\n" +
                "  'sink.buffer-flush.max-rows' = '1000',\n" +
                "  'zookeeper.quorum' = '10.201.7.186:2181,10.201.7.187:2181,10.201.7.188:2181'\n" +
                ")";
        List<CreateTableParser.SqlParserResult> result = CreateTableParser.newInstance().parseSql(sql2);
        for(CreateTableParser.SqlParserResult sqlParserResult : result){
            Meta meta = new Meta();
            meta.setTableName(sqlParserResult.getTableName());
            Object component = sqlParserResult.getPropMap().get("'connector'");
            if(component != null){
                meta.setComponent(component.toString());
            }
            String[] fieldsInfoStr = sqlParserResult.getFieldsInfoStr().split(",");
            String formateFields = Arrays.stream(fieldsInfoStr).map((field)-> field.trim()).collect(Collectors.joining(","));
            meta.setFieldInfo(formateFields);
            meta.setCreateTableStatement(sqlParserResult.getCreateTableStatement());
            meta.setProp(JSON.toJSONString(sqlParserResult.getPropMap()));
            meta.setDatabaseName("default");
            System.out.println(meta);
        }
    }
}
